package com.geeguo.ebuilder.client.base.localcache.listener;

import com.geeguo.ebuilder.client.base.core.constants.MQConsumerGroupConstants;
import com.geeguo.ebuilder.client.base.core.constants.MQTopicConstants;
import com.geeguo.ebuilder.core.common.utils.GsonUtils;
import com.geeguo.ebuilder.core.common.utils.worker.WorkerIdHelper;
import com.geeguo.ebuilder.core.localcache.event.LocalCacheChangeEventMessage;
import com.geeguo.ebuilder.core.localcache.service.LocalCacheService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RocketMQMessageListener(topic = MQTopicConstants.CLIENT_LOCAL_CACHE_CHANGE, consumerGroup = MQConsumerGroupConstants.CLIENT_LOCAL_CACHE_CHANGE, messageModel = MessageModel.BROADCASTING)
public class LocalCacheChangeEventListener implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private LocalCacheService<Object> localCacheService;

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setInstanceName(String.valueOf(WorkerIdHelper.getWorkerId()));
    }

    @Override
    public void onMessage(String message) {
        if (message != null) {
            LocalCacheChangeEventMessage eventMessage = GsonUtils.fromJson(message, LocalCacheChangeEventMessage.class);
            if (eventMessage != null && eventMessage.getWorkerId() != WorkerIdHelper.getWorkerId()) {
                localCacheService.removeAllWithoutEvent(eventMessage.getCacheName(), eventMessage.getKeys());
            }
        }
    }
}
